package queue

import (
	"context"
	pq "github.com/emirpasic/gods/queues/priorityqueue"
	"github.com/emirpasic/gods/utils"
	"github.com/kataras/golog"
	"sdn-executor/utils/erros"
	"sync"
	"time"
)

type ClosePolicy int

const (
	ClosePolicyClose ClosePolicy = iota
	ClosePolicyCloseNow
)

type Element interface {
	Priority() int
}

// BlockingPriorityQueue
// maxSize <= 0，表示无界队列；maxSize > 0，表示有界队列
type BlockingPriorityQueue[E Element] struct {
	que         *pq.Queue
	maxSize     int
	InputChan   chan E
	OutputChan  chan E
	context     context.Context
	closePolicy ClosePolicy
	writeClosed bool
	sync.Once
}

func NewBlockingPriorityQueue[E Element](c context.Context, maxSize int, policy ClosePolicy) (*BlockingPriorityQueue[E], *erros.Error) {
	if c == nil {
		return nil, erros.New(erros.CodeInvalidQueryParam, "InvalidParameter")
	}

	return &BlockingPriorityQueue[E]{
		que:         pq.NewWith(byPriority),
		maxSize:     maxSize,
		InputChan:   make(chan E),
		OutputChan:  make(chan E),
		context:     c,
		closePolicy: policy,
	}, nil
}

func NewAndStartBlockingPriorityQueue[E Element](c context.Context, maxSize int, policy ClosePolicy) (*BlockingPriorityQueue[E], *erros.Error) {
	q, err := NewBlockingPriorityQueue[E](c, maxSize, policy)
	if err != nil {
		return q, err
	}
	q.Start()
	return q, nil
}

func (q *BlockingPriorityQueue[E]) Start() {
	if q != nil {
		go q.Do(q.run)
	}
}

// Push 阻塞型写入元素，可能触发panic
func (q *BlockingPriorityQueue[E]) Push(e E) (err *erros.Error) {
	defer func() {
		if r := recover(); r != nil {
			err = erros.New(erros.CodeResourceNotExisted, "write closed channel panic")
		}
	}()

	q.InputChan <- e

	err = nil
	return err
}

// PushWithTimeout 超时，可能触发panic
func (q *BlockingPriorityQueue[E]) PushWithTimeout(e E, timeout time.Duration) (err *erros.Error) {
	defer func() {
		if r := recover(); r != nil {
			err = erros.New(erros.CodeResourceNotExisted, "write closed channel panic")
		}
	}()

	select {
	case q.InputChan <- e:
		err = nil
		return err
	case <-time.After(timeout):
		return erros.New(erros.CodeSystemBusy, "system is busy, try later")
	}

}

// Pop 阻塞型读取元素
func (q *BlockingPriorityQueue[E]) Pop() (E, *erros.Error) {
	e, ok := <-q.OutputChan
	if ok {
		return e, nil
	}
	return e, erros.New(erros.CodeResourceNotExisted, "closed")
}

// PopWithTimeout 超时
func (q *BlockingPriorityQueue[E]) PopWithTimeout(timeout time.Duration) (E, *erros.Error) {
	select {
	case e, ok := <-q.OutputChan:
		if ok {
			return e, nil
		}
		return e, erros.New(erros.CodeResourceNotExisted, "closed")
	case <-time.After(timeout):
		var e E
		return e, erros.New(erros.CodeSystemBusy, "timeout")
	}
}

func (q *BlockingPriorityQueue[E]) run() {
	defer func() {
		if r := recover(); r != nil {
			golog.Errorf("Recovered. err: %+v", r)
		}
	}()

	for {
		if q.closePolicy == ClosePolicyClose && q.writeClosed {
			// 已关闭
			if q.que.Empty() {
				// 关闭输出，然后结束
				close(q.OutputChan)
				return
			} else {
				// 只读队列消息
				select {
				case q.OutputChan <- func() E {
					v, _ := q.que.Peek()
					return v.(E)
				}():
					q.que.Dequeue()
				}
			}
		} else {
			// 正常运行
			if terminated := q.processWhileRunning(); terminated {
				return
			}
		}
	}
}

func (q *BlockingPriorityQueue[E]) processWhileRunning() bool {
	if q.que.Empty() {
		// 队列为空，只能写入元素
		select {
		case e := <-q.InputChan:
			q.que.Enqueue(e)
		case <-q.context.Done():
			return q.close()
		}
	} else if q.maxSize > 0 && q.que.Size() >= q.maxSize {
		// 有界队列，且队列已满，不能写入元素，只能读取元素
		select {
		case q.OutputChan <- func() E {
			v, _ := q.que.Peek()
			return v.(E)
		}():
			q.que.Dequeue()
		case <-q.context.Done():
			return q.close()
		}
	} else {
		// 队列非空；有界队列非满或无界队列，可读可写元素
		select {
		case e, ok := <-q.InputChan:
			if ok {
				q.que.Enqueue(e)
			}
		case q.OutputChan <- func() E {
			v, _ := q.que.Peek()
			return v.(E)
		}():
			q.que.Dequeue()
		case <-q.context.Done():
			return q.close()
		}
	}
	return false
}

// Close 关闭队列, true: 立即退出；false：队列读空后退出
func (q *BlockingPriorityQueue[E]) close() bool {
	if q.closePolicy == ClosePolicyCloseNow {
		close(q.InputChan)
		close(q.OutputChan)
		q.que.Clear()
		return true
	} else {
		close(q.InputChan)
		q.writeClosed = true
		return false
	}
}

func byPriority(a interface{}, b interface{}) int {
	pa := a.(Element).Priority()
	pb := b.(Element).Priority()
	return utils.IntComparator(pa, pb) // "-" descending order
}
